/*
 * BlockingQueue.c
 *
 *  Created on: 2015年6月13日
 *      Author: vincent
 */

#include "LinkedList.h"
#include "BlockingQueue.h"

#include <stdlib.h>
#include <memory.h>
#include <pthread.h>

/**
 * 构造阻塞队列
 */
BLOCKING_QUEUE* blocking_queue_construct(unsigned int capacity)
{
	BLOCKING_QUEUE* blocking_queue = NULL;
	blocking_queue = malloc(sizeof(BLOCKING_QUEUE));
	if(blocking_queue == NULL) return NULL;
	memset(blocking_queue,  0, sizeof(BLOCKING_QUEUE));
	blocking_queue->mutex_for_put = malloc(sizeof(pthread_mutex_t));
	blocking_queue->mutex_for_put_full_wait =  malloc(sizeof(pthread_mutex_t));
	blocking_queue->cond_for_put_full_wait = malloc(sizeof(pthread_cond_t));
	blocking_queue->mutex_for_take = malloc(sizeof(pthread_mutex_t));
	blocking_queue->mutex_for_take_empty_wait =  malloc(sizeof(pthread_mutex_t));
	blocking_queue->cond_for_take_empty_wait = malloc(sizeof(pthread_cond_t));
	pthread_mutex_init(blocking_queue->mutex_for_put, NULL);
	pthread_mutex_init(blocking_queue->mutex_for_put_full_wait, NULL);
	pthread_cond_init(blocking_queue->cond_for_put_full_wait, NULL);
	pthread_mutex_init(blocking_queue->mutex_for_take, NULL);
	pthread_mutex_init(blocking_queue->mutex_for_take_empty_wait, NULL);
	pthread_cond_init(blocking_queue->cond_for_take_empty_wait, NULL);
	blocking_queue->item_list = linked_list_construct();
	if(capacity > 0)
	{
		blocking_queue->capacity = capacity;
	}
	return blocking_queue;
}

/**
 * 销毁阻塞队列
 */
int blocking_queue_destroy(BLOCKING_QUEUE* blocking_queue)
{
	blocking_queue->is_destroy = 1;
	linked_list_destroy(blocking_queue->item_list);
	pthread_mutex_destroy(blocking_queue->mutex_for_put);
	pthread_mutex_destroy(blocking_queue->mutex_for_put_full_wait);
	pthread_cond_destroy(blocking_queue->cond_for_put_full_wait);
	pthread_mutex_destroy(blocking_queue->mutex_for_take);
	pthread_mutex_destroy(blocking_queue->mutex_for_take_empty_wait);
	pthread_cond_destroy(blocking_queue->cond_for_take_empty_wait);
	free(blocking_queue->mutex_for_put);
	free(blocking_queue->mutex_for_put_full_wait);
	free(blocking_queue->cond_for_put_full_wait);
	free(blocking_queue->mutex_for_take);
	free(blocking_queue->mutex_for_take_empty_wait);
	free(blocking_queue->cond_for_take_empty_wait);
	free(blocking_queue);
	return 0;
}

/**
 * 从队列中取出元素
 */
int blocking_queue_take(BLOCKING_QUEUE* blocking_queue, void** target)
{
	int err = 0, is_polled = 0;
	while(1)
	{
		if(blocking_queue->item_list->length > 0)
		{
			err = pthread_mutex_lock(blocking_queue->mutex_for_take);
			if(!err && (blocking_queue->item_list->length > 0))
			{
				err = linked_list_poll_first(blocking_queue->item_list, target);
				is_polled = 1;
			}
			pthread_cond_signal(blocking_queue->cond_for_put_full_wait);
			err = pthread_mutex_unlock(blocking_queue->mutex_for_take);
			if(is_polled)
			{
				return err;
			}
		}

		err = pthread_mutex_lock(blocking_queue->mutex_for_take_empty_wait);
		err = pthread_cond_wait(blocking_queue->cond_for_take_empty_wait,
				blocking_queue->mutex_for_take_empty_wait);
		err = pthread_mutex_unlock(blocking_queue->mutex_for_take_empty_wait);
		if(blocking_queue->is_destroy)
		{
			return 1;
		}
	}
	return err;
}

/**
 * 向队列中放入元素
 */
int blocking_queue_put(BLOCKING_QUEUE* blocking_queue, void* target)
{
	int err = 0, is_offered = 0;
	while(1)
	{
		if(blocking_queue->capacity > 0)
		{
			if(blocking_queue->item_list->length < blocking_queue->capacity)
			{
				err = pthread_mutex_lock(blocking_queue->mutex_for_put);
				if(blocking_queue->item_list->length < blocking_queue->capacity)
				{
					linked_list_offer_last(blocking_queue->item_list, target);
					is_offered = 1;
				}
				pthread_cond_signal(blocking_queue->cond_for_take_empty_wait);
				err = pthread_mutex_unlock(blocking_queue->mutex_for_put);
				if(is_offered)
				{
					return err;
				}
			}

			err = pthread_mutex_lock(blocking_queue->mutex_for_put_full_wait);
			err = pthread_cond_wait(blocking_queue->cond_for_put_full_wait,
					blocking_queue->mutex_for_put_full_wait);
			err = pthread_mutex_unlock(blocking_queue->mutex_for_put_full_wait);
			if(blocking_queue->is_destroy)
			{
				return 1;
			}
		}
	}
}


